package org.example;



import lombok.SneakyThrows;
import lombok.val;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Author:   susq
 * Date:     2019-08-17 11:07
 */
public class TestPool {
    public static void main(String[] args) {
        ThreadPool pool=new ThreadPool(2,1,TimeUnit.SECONDS,10,((queue, task) -> {
            //策略模式
            System.out.println("reject");

        }));

        for (int i=0;i<20;i++){
            int finalI = i;
            pool.exec(() -> {
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println("线程"+Thread.currentThread().getName()+"执行了"+"   "+ finalI);

            });
        }
    }

}

interface RejectPolicy<T>{
    void reject(BlockingQueue<T> queue,T task);
}

class ThreadPool{
    //任务队列
    private BlockingQueue<Runnable> taskQueue;

    //线程集合
    private HashSet<Worker> workers=new HashSet<Worker>();

    //核心线程数
    private int coreSize;

    //获取任务的超时时间
    private long timeout;

    private TimeUnit timeUnit;

    private int capacity;

    private RejectPolicy<Runnable> rejectPolicy;
    public ThreadPool(int coreSize,long timeout,TimeUnit timeUnit,int capacity,RejectPolicy<Runnable> rejectPolicy){
        this.coreSize=coreSize;
        this.timeout=timeout;
        this.timeUnit=timeUnit;
        this.taskQueue= new BlockingQueue<>(capacity);
        this.rejectPolicy=rejectPolicy;
    }


    class Worker extends Thread{
        Runnable task;
        public Worker(Runnable task) {
            this.task=task;
        }

        @SneakyThrows
        @Override
        public void run() {
            while (task!=null||(task=taskQueue.takeWithCancel(timeout,timeUnit))!=null){
                try {
                    System.out.printf("执行任务 :%s\n",task);
                  task.run();
              }catch (Exception e){
                 e.printStackTrace();
              }finally {
                    task=null;
                }
            }
            synchronized (workers){
                System.out.printf("移除线程:%s\n",this);
             workers.remove(this);

            }



        }
    }

    public void exec(Runnable task){
        synchronized (workers){
            if(workers.size()<coreSize){

                Worker worker=new Worker(task);
                workers.add(worker);
                System.out.printf("新增线程 :%s   %s\n",worker,task);
                worker.start();
            }else {
                try {
                    System.out.printf("线程数达到上限，把任务加入任务队列%s\n",task);
                    //策略
                    //死等
                    //超时等待
                    //调用者放弃
                    //调用者抛出异常
                    //调用者自己执行
                    taskQueue.tryPut(rejectPolicy,task);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }
        }

}
}

//任务阻塞队列
class BlockingQueue<T>{
    private Deque<T> deque=new ArrayDeque<>();

    private ReentrantLock lock=new ReentrantLock();

    //生产者条件变量
    private Condition fullWaitSet= lock.newCondition();


    //消费者条件变量
    private Condition emptyWaitSet= lock.newCondition();

    //容量
    private int capcity;

    public BlockingQueue(int capcity){
        this.capcity=capcity;
    }
    //阻塞获取
    public T take() throws InterruptedException {
        lock.lock();
        try{
            while (deque.isEmpty()){
                emptyWaitSet.await();
            }
            fullWaitSet.signal();
            return deque.removeFirst();

        }finally {
            lock.unlock();
        }

    }

    //阻塞获取 带超时时间
    public T takeWithCancel(long timeout, TimeUnit unit) throws InterruptedException {
        lock.lock();
        try{
            long nanos=unit.toNanos(timeout);
            while (deque.isEmpty()){
                if(nanos<=0){
                    return null;
                }
                nanos=emptyWaitSet.awaitNanos(nanos);
            }
            fullWaitSet.signal();
            return deque.removeFirst();

        }  finally {
            lock.unlock();
        }

    }


    //阻塞添加
    public void put(T element) throws InterruptedException {
        lock.lock();
        try{
            while (deque.size()==capcity){
                System.out.printf("陷入等待%s\n",element);

                fullWaitSet.await();
            }
            System.out.printf("加入任务队列%s\n",element);
            deque.addLast(element);
            emptyWaitSet.signal();

        }finally {
            lock.unlock();
        }
    }


    //带超时时间的阻塞添加
    public boolean offer(T element,long timeout, TimeUnit unit) throws InterruptedException {
        lock.lock();
        try{
            long nanos=unit.toNanos(timeout);
            while (deque.size()==capcity){
                if (nanos<=0){
                    System.out.printf("等待超时%s\n",element);
                    return false;
                }
                System.out.printf("陷入等待%s\n",element);

                nanos=fullWaitSet.awaitNanos(nanos);
            }
            System.out.printf("加入任务队列%s\n",element);
            deque.addLast(element);
            emptyWaitSet.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    public int size(){
        lock.lock();
        try{
            return deque.size();
        }finally {
            lock.unlock();

        }
    }

    public void tryPut(RejectPolicy<T> rejectPolicy, T task) {
      lock.lock();
      try {
          if (deque.size()==capcity){
              rejectPolicy.reject(this,task);
          }else {
              System.out.printf("加入任务队列%s\n",task);
              deque.addLast(task);
              emptyWaitSet.signal();
          }
      }finally {
          lock.unlock();
      }
    }
}